
/*
 * Copyright © 2021 https://www.cestc.cn/ All rights reserved.
 */

package com.zx.learn.flink.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Random;
import java.util.UUID;

/**
 * 几种 SourceFunction
 * SourceFunction:非并行数据源(并行度只能=1)
 * RichSourceFunction:多功能非并行数据源(并行度只能=1)
 * ParallelSourceFunction:并行数据源(并行度能够>=1)
 * RichParallelSourceFunction:多功能并行数据源(并行度能够>=1)–后续学习的Kafka数据源使用的
 * <p>
 * 每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
 * 要求:
 * - 随机生成订单ID(UUID)
 * - 随机生成用户ID(0-2)
 * - 随机生成订单金额(0-100)
 * - 时间戳为当前系统时间
 */
public class CustomSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //自定义 source
        Random rn = new Random();
        DataStreamSource<Order> source = env.addSource(new ParallelSourceFunction<Order>() {
            boolean flag = true;

            //创建一个 标记
            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                int loop = 0;
                while (flag) {
                    //随机生成订单ID(UUID)
                    String oid = UUID.randomUUID().toString();
                    //随机生成用户ID(0-2)
                    int uid = rn.nextInt(3);
                    //随机生成订单金额(0-100)
                    int money = rn.nextInt(101);
                    //时间戳为当前系统时间
                    long timestamp = System.currentTimeMillis();
                    //将数据封装成 Order 收集数据
                    ctx.collect(new Order(oid, uid, money, timestamp));
                    //每一秒休息一次
                    Thread.sleep(1000);
                    if (loop++ >= 10) {
                        break;
                    }
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        }).setParallelism(1);
        //打印输出
        source.print();
        env.execute();
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    static class Order {
        private String uuid;
        private int uid;
        private int money;
        private Long timestamp;
    }

}
